package com.yangji.rxjavatest.rx.flowable;

import android.os.Handler;
import android.os.Message;

import com.yangji.rxjavatest.Log;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.FlowableEmitter;
import io.reactivex.FlowableOnSubscribe;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.schedulers.Schedulers;

/**
 * http://www.jianshu.com/p/9b1304435564
 * Flowable 升级
 */
public class RxFlowable02 {

    private static final String TAG = "MainActivity";

    private Subscription mSubscription;

    public RxFlowable02() {
        Flowable<Integer> flowable = Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
                //这里只能有128个 超过128就会抛出异常 如果下游能消费过来 就不会出错 缓冲区之恩给你有128
                //原因是BackpressureStrategy.ERROR 详细请看下一张
                for (int i = 0; i < 129; i++) {
                    Log.d(TAG, "emit " + i);
                    emitter.onNext(i);
                }
                emitter.onComplete();
            }
        }, BackpressureStrategy.ERROR);


        Subscriber<Integer> subscriber = new Subscriber<Integer>() {
            @Override
            public void onSubscribe(Subscription s) {
                Log.d(TAG, "onSubscribe");
                //注意这句代码 这句话暂时注释后onNext无法收到消息（同步抛出异常，异步线程不会抛出异常）代表下游需要获取多少数据
                //或者根据上游发送事件的数量来request就行了, 比如这里request(3)就可以了.
//                s.request(Long.MAX_VALUE);
                mSubscription = s;
            }

            @Override
            public void onNext(Integer integer) {
                Log.d(TAG, "onNext: " + integer);

            }

            @Override
            public void onError(Throwable t) {
                Log.w(TAG, "onError: ", t);
            }

            @Override
            public void onComplete() {
                Log.d(TAG, "onComplete");
            }

        };
        flowable
                .subscribeOn(Schedulers.io())//加入这两句Subscription.request可以注释 不会抛出异常 但是依然不能收到数据
                .observeOn(AndroidSchedulers.mainThread())//加入这两句Subscription.request可以注释 但是依然不能收到数据
                .subscribe(subscriber);
        start();

    }


    /**
     * 模拟用户点击 每次点击获取2条数据
     */
    private void start() {
        Handler handler = new Handler() {
            @Override
            public void handleMessage(Message msg) {
                super.handleMessage(msg);
                mSubscription.request(100);//保持缓冲区只有128位 129会抛出异常
                sendEmptyMessageDelayed(1, 2000);
            }
        };
        handler.sendEmptyMessage(1);
    }


}
